package com.xuecheng.media.rocketmq;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.Gson;
import com.google.gson.JsonSyntaxException;
import com.j256.simplemagic.logger.Logger;
import com.j256.simplemagic.logger.LoggerFactory;
import com.xuecheng.content.model.message.VideoProcessMessage;
import com.xuecheng.media.jobhandler.ThreadPoolMessageVideoTask;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

@Component
@RocketMQMessageListener(
        topic = "VIDEO_PROCESS_TOPIC",
        consumerGroup = "${rocketmq.consumer.group:VIDEO_PROCESS_CONSUMER_GROUP}",
        consumeThreadMax = 20, // 增加消费线程
        consumeTimeout = 5 // 超时时间(分钟)
)
public class VideoProcessingListener implements RocketMQListener<MessageExt> {

    private static final Logger LOG = LoggerFactory.getLogger(VideoProcessingListener.class);

    @Autowired
    private ThreadPoolMessageVideoTask threadPoolMessageVideoTask;

    @Autowired
    private ObjectMapper objectMapper;

    private static final Gson gson = new Gson(); // 备选解析器

    @Override
    public void onMessage(MessageExt messageExt) {
        final String msgId = messageExt.getMsgId();
        try {
            LOG.info("===== 开始处理消息 [{}] =====", msgId);

            // 1. 获取消息体
            String messageBody = new String(messageExt.getBody(), StandardCharsets.UTF_8);
            LOG.debug("原始消息体: {}", messageBody);

            // 2. 解析消息
            VideoProcessMessage message = parseMessage(messageBody);

            // 3. 验证消息
            validateMessage(message);

            LOG.info("成功解析消息 | 文件ID: {}", message.getFileId());

            // 4. 委托处理
            threadPoolMessageVideoTask.handleProcessingRequest(message);

            LOG.info("===== 消息处理完成 [{}] =====", msgId);
        } catch (Exception e) {
            LOG.error("消息处理失败 [{}]", msgId, e);
            // 根据重试策略决定是否重试
            if (messageExt.getReconsumeTimes() < 3) {
                throw new RuntimeException("需要重试", e);
            }
        }
    }

    private VideoProcessMessage parseMessage(String json) {
        try {
            // 首选使用ObjectMapper
            return objectMapper.readValue(json, VideoProcessMessage.class);
        } catch (JsonProcessingException e) {
            LOG.warn("JSON解析失败，尝试Gson解析", e);
            try {
                // 备选使用Gson
                return gson.fromJson(json, VideoProcessMessage.class);
            } catch (JsonSyntaxException ex) {
                throw new IllegalArgumentException("无法解析消息", ex);
            }
        }
    }

    private void validateMessage(VideoProcessMessage message) {
        if (message == null) {
            throw new IllegalArgumentException("消息为空");
        }

        List<String> missingFields = new ArrayList<>();
        if (message.getFileId() == null || message.getFileId().isEmpty())
            missingFields.add("fileId");
        if (message.getOriginalUrl() == null || message.getOriginalUrl().isEmpty())
            missingFields.add("originalUrl");
        if (message.getFilename() == null || message.getFilename().isEmpty())
            missingFields.add("filename");
        if (message.getBucket() == null || message.getBucket().isEmpty())
            missingFields.add("bucket");

        if (!missingFields.isEmpty()) {
            throw new IllegalArgumentException("缺少必要字段: " + missingFields);
        }
    }
}

